package org.com.lcuiot.config;


import org.com.lcuiot.constant.LcuIotConst;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;


import javax.annotation.Resource;
import java.time.Duration;


/**
 * @author gaoge
 * @since 2022/10/27 14:54
 */
@Configuration
public class RedisStreamConfig implements ApplicationRunner, DisposableBean {
    private static final Logger logger = LoggerFactory.getLogger(IotMsgListener.class);

    private final IotMsgListener iotMsgListener;
    private final StringRedisTemplate redisTemplate;
    @Resource
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    private StreamMessageListenerContainer listenerContainer;

    @Autowired
    public RedisStreamConfig(IotMsgListener iotMsgListener, StringRedisTemplate redisTemplate) {
        this.iotMsgListener = iotMsgListener;
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void destroy() throws Exception {
        this.listenerContainer.stop();
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (Boolean.TRUE.equals(redisTemplate.hasKey(LcuIotConst.IOT_STREAM))) {
            StreamInfo.XInfoGroups groups = redisTemplate.opsForStream().groups(LcuIotConst.IOT_STREAM);
            if (groups.isEmpty()) {
                redisTemplate.opsForStream().createGroup(LcuIotConst.IOT_STREAM, LcuIotConst.IOT_STREAM_GROUP);
            }
        } else {
            redisTemplate.opsForStream().createGroup(LcuIotConst.IOT_STREAM, LcuIotConst.IOT_STREAM_GROUP);
        }

        RedisConnectionFactory factory;
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ?> streamMessageListenerContainerOptions;
        factory = redisTemplate.getConnectionFactory();
        streamMessageListenerContainerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次性最多拉取多少条消息
                .batchSize(10)
                // 执行消息轮询的执行器
                .executor(this.threadPoolTaskExecutor)
                // 消息消费异常的handler
                .errorHandler(t -> {
                    t.printStackTrace();
                    logger.error("[MQ handler exception] {}", t.getMessage());
                })
                // 超时时间，设置为0，表示不超时（超时后会抛出异常）
                .pollTimeout(Duration.ZERO)
                // 序列化器
                .serializer(new StringRedisSerializer())
                .build();

        this.listenerContainer = StreamMessageListenerContainer.create(factory, streamMessageListenerContainerOptions);
        listenerContainer.start();

        listenerContainer.receive(
                Consumer.from(LcuIotConst.IOT_STREAM_GROUP,LcuIotConst.IOT_CONSUMER_NAME),
                StreamOffset.create(LcuIotConst.IOT_STREAM, ReadOffset.lastConsumed()),
                iotMsgListener);
    }
}
